import org.apache.flink.table.functions.ScalarFunction;


/*
 * We can put frequently used procedures in user-defined functions.
 *
 * <p>It is possible to call third-party libraries here as well.
 */


public class AddressNormalizer extends ScalarFunction
{

    // the 'eval()' method defines input and output types (reflectively extracted)
    // and contains the runtime logic
    public String eval(String street, String zipCode, String city)
    {
        return normalize(street) + ", " + normalize(zipCode) + ", " + normalize(city);
    }

    private String normalize(String s) {
        return s.toUpperCase().replaceAll("\\W", " ").replaceAll("\\s+", " ").trim();
    }
}